package client_core

import (
	"context"
	"errors"
	"github.com/chicken-team-outside/chicken_transmission/client_core/service"
	"github.com/chicken-team-outside/chicken_transmission/model"
	"go.uber.org/zap"
	"sync"
	"time"
)

type TaskMgrOpType uint8

const (
	TaskMgrOpTypeAddTask TaskMgrOpType = iota
	TaskMgrOpTypePauseTask
	TaskMgrOpTypeResumeTask
	TaskMgrOpTypeDeleteTask
	TaskMgrOpTypeUpdateStatus
)

var (
	ErrShutdown              = errors.New("task runner shutdown")
	ErrTaskFinished          = errors.New("task finished")
	ErrTaskPaused            = errors.New("task paused")
	ErrOccurredPanic         = errors.New("occurred panic")
	ErrSFSClientNotAvailable = errors.New("sfs client not available")
	ErrTaskInProgress        = errors.New("existing tasks in progress")
	canResumeStatus          = []model.TaskStatus{model.TaskStatus_Paused, model.TaskStatus_Failed}
	canPauseStatus           = []model.TaskStatus{model.TaskStatus_Waiting, model.TaskStatus_Running}
	canDeleteStatus          = []model.TaskStatus{model.TaskStatus_Failed, model.TaskStatus_Finished, model.TaskStatus_Paused}
)

type TaskRunner struct {
	runCB            RunTaskCB
	taskType         model.TaskType
	ctx              context.Context
	cancel           context.CancelCauseFunc
	tasks            map[int64]*TaskWarp
	taskLock         sync.RWMutex
	taskProducerLock sync.Mutex
	taskProducerCon  sync.Cond
	wg               sync.WaitGroup
}

func (r *TaskRunner) queryWaitTask() (*TaskWarp, error) {
	task, err := service.GetOneWaitingTask(r.taskType)
	if err != nil {
		return nil, err
	}
	if task != nil {
		pt := &TaskWarp{
			info:   task,
			runner: r,
		}
		pt.ctx, pt.cancel = context.WithCancelCause(r.ctx)
		err = service.UpdateStatus(task, model.TaskStatus_Running, "")
		if err != nil {
			return nil, err
		}
		pt.SetTotal(task.Total)
		pt.SetCompleted(task.Completed)
		sendTaskStatus(task.TaskType, task.Id, model.TaskStatus_Running, "", task.Completed)
		r.taskLock.Lock()
		r.tasks[task.Id] = pt
		r.taskLock.Unlock()
		return pt, nil
	}
	return nil, nil
}

func (r *TaskRunner) taskProducer() *TaskWarp {
	r.taskProducerLock.Lock()
	defer r.taskProducerLock.Unlock()
	for r.ctx.Err() == nil {
		task, err := r.queryWaitTask()
		if err != nil {
			zap.L().Error("failed to get task from db", zap.Error(err), zap.Any("type", r.taskType))
			r.cancel(err)
			return nil
		}
		if task == nil && r.ctx.Err() == nil {
			r.taskProducerCon.Wait()
		} else {
			return task
		}
	}
	return nil
}

func (r *TaskRunner) runTask(task *TaskWarp) {
	defer func() {
		r.taskLock.Lock()
		defer r.taskLock.Unlock()
		delete(r.tasks, task.info.Id)
	}()
	task.run()
}

func (r *TaskRunner) taskConsumer() {
	defer r.shutdown()
	defer r.wg.Done()
	for {
		task := r.taskProducer()
		if task == nil {
			return
		}
		r.runTask(task)
	}
}

func (r *TaskRunner) notifyTask() {
	r.taskProducerLock.Lock()
	defer r.taskProducerLock.Unlock()
	r.taskProducerCon.Broadcast()
}

func (r *TaskRunner) AddTasks(task []*model.Task) (err error) {
	if len(task) > 0 {
		err = service.AddTasks(task)
		if err == nil {
			r.notifyTask()
			sendTaskStatusChange(r.taskType, TaskMgrOpTypeAddTask, task)
		}
	}
	return
}

func (r *TaskRunner) Startup(maxWorkers int) {
	r.wg.Add(maxWorkers)
	for i := 0; i < maxWorkers; i++ {
		go r.taskConsumer()
	}
}

func (r *TaskRunner) shutdown() {
	r.cancel(ErrShutdown)
	r.notifyTask()
}

func (r *TaskRunner) opWithLock(withTaskLock bool, f func() error) error {
	if withTaskLock {
		r.taskProducerLock.Lock()
		defer r.taskProducerLock.Unlock()
	}
	return f()
}

func (r *TaskRunner) Shutdown() {
	r.shutdown()
	r.wg.Wait()
}

func (r *TaskRunner) GetRunningTaskStatus() (result []*TaskProgress) {
	r.taskLock.RLock()
	defer r.taskLock.RUnlock()
	result = make([]*TaskProgress, 0, len(r.tasks))
	for _, pt := range r.tasks {
		result = append(result, &TaskProgress{
			Id:        pt.info.Id,
			Completed: pt.completed.Load(),
			UpdateAt:  time.Now().UnixMilli(),
		})
	}
	return
}

func NewTaskRunner(ctx context.Context, taskType model.TaskType, runCB RunTaskCB) *TaskRunner {
	r := &TaskRunner{
		runCB:    runCB,
		taskType: taskType,
		tasks:    make(map[int64]*TaskWarp),
	}
	r.ctx, r.cancel = context.WithCancelCause(ctx)
	r.taskProducerCon.L = &r.taskProducerLock
	return r
}

func sendTaskStatus(taskType model.TaskType, id int64, status model.TaskStatus, msg string, completed int64) {
	sendEventData("task-status-change", taskType, TaskMgrOpTypeUpdateStatus, []interface{}{id, status, msg, completed})
}

func sendTaskStatusChangeByIds(taskType model.TaskType, opType TaskMgrOpType, ids []int64) {
	sendEventData("task-status-change", taskType, opType, ids)
}

func sendTaskStatusChange(taskType model.TaskType, opType TaskMgrOpType, task []*model.Task) {
	sendEventData("task-status-change", taskType, opType, task)
}
